<!DOCTYPE html>



  


<html class="theme-next muse use-motion" lang="zh-Hans">
<head>
  <meta charset="UTF-8"/>
<meta http-equiv="X-UA-Compatible" content="IE=edge" />
<meta name="viewport" content="width=device-width, initial-scale=1, maximum-scale=1"/>
<meta name="theme-color" content="#222">









<meta http-equiv="Cache-Control" content="no-transform" />
<meta http-equiv="Cache-Control" content="no-siteapp" />
















  
  
  <link href="/lib/fancybox/source/jquery.fancybox.css?v=2.1.5" rel="stylesheet" type="text/css" />







<link href="/lib/font-awesome/css/font-awesome.min.css?v=4.6.2" rel="stylesheet" type="text/css" />

<link href="/css/main.css?v=5.1.4" rel="stylesheet" type="text/css" />


  <link rel="apple-touch-icon" sizes="180x180" href="/images/apple-touch-icon-next.png?v=5.1.4">


  <link rel="icon" type="image/png" sizes="32x32" href="/images/favicon-32x32-next.png?v=5.1.4">


  <link rel="icon" type="image/png" sizes="16x16" href="/images/favicon-16x16-next.png?v=5.1.4">


  <link rel="mask-icon" href="/images/logo.svg?v=5.1.4" color="#222">





  <meta name="keywords" content="Hexo, NexT" />










<meta name="description" content="本文沿着《RocketMQ消息轨迹-设计篇》的思路，从如下3个方面对其源码进行解读：  发送消息轨迹 消息轨迹格式 存储消息轨迹数据  1、发送消息轨迹流程首先我们来看一下在消息发送端如何启用消息轨迹，示例代码如下：">
<meta property="og:type" content="article">
<meta property="og:title" content="源码分析RocketMQ消息轨迹">
<meta property="og:url" content="https://www.codingw.net/posts/e8f03b64.html">
<meta property="og:site_name" content="中间件兴趣圈">
<meta property="og:description" content="本文沿着《RocketMQ消息轨迹-设计篇》的思路，从如下3个方面对其源码进行解读：  发送消息轨迹 消息轨迹格式 存储消息轨迹数据  1、发送消息轨迹流程首先我们来看一下在消息发送端如何启用消息轨迹，示例代码如下：">
<meta property="og:locale">
<meta property="og:image" content="https://img-blog.csdnimg.cn/20190803203518449.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ByZXN0aWdlZGluZw==,size_16,color_FFFFFF,t_70">
<meta property="og:image" content="https://img-blog.csdnimg.cn/20190803204514547.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ByZXN0aWdlZGluZw==,size_16,color_FFFFFF,t_70">
<meta property="article:published_time" content="2020-12-15T14:01:35.000Z">
<meta property="article:modified_time" content="2021-04-26T12:11:31.146Z">
<meta property="article:author" content="中间件兴趣圈">
<meta property="article:tag" content="中间件">
<meta name="twitter:card" content="summary">
<meta name="twitter:image" content="https://img-blog.csdnimg.cn/20190803203518449.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ByZXN0aWdlZGluZw==,size_16,color_FFFFFF,t_70">



<script type="text/javascript" id="hexo.configurations">
  var NexT = window.NexT || {};
  var CONFIG = {
    root: '',
    scheme: 'Muse',
    version: '5.1.4',
    sidebar: {"position":"left","display":"post","offset":12,"b2t":false,"scrollpercent":false,"onmobile":false},
    fancybox: true,
    tabs: true,
    motion: {"enable":true,"async":false,"transition":{"post_block":"fadeIn","post_header":"slideDownIn","post_body":"slideDownIn","coll_header":"slideLeftIn","sidebar":"slideUpIn"}},
    duoshuo: {
      userId: '0',
      author: '博主'
    },
    algolia: {
      applicationID: '',
      apiKey: '',
      indexName: '',
      hits: {"per_page":10},
      labels: {"input_placeholder":"Search for Posts","hits_empty":"We didn't find any results for the search: ${query}","hits_stats":"${hits} results found in ${time} ms"}
    }
  };
</script>



  <link rel="canonical" href="https://www.codingw.net/posts/e8f03b64.html"/>





  <title>源码分析RocketMQ消息轨迹 | 中间件兴趣圈</title>
  








<meta name="generator" content="Hexo 5.4.0"></head>

<body itemscope itemtype="http://schema.org/WebPage" lang="zh-Hans">

  
  
    
  

  <div class="container sidebar-position-left page-post-detail">
    <div class="headband"></div>

    <header id="header" class="header" itemscope itemtype="http://schema.org/WPHeader">
      <div class="header-inner"><div class="site-brand-wrapper">
  <div class="site-meta ">
    

    <div class="custom-logo-site-title">
      <a href="/"  class="brand" rel="start">
        <span class="logo-line-before"><i></i></span>
        <span class="site-title">中间件兴趣圈</span>
        <span class="logo-line-after"><i></i></span>
      </a>
    </div>
      
        <p class="site-subtitle">微信搜『中间件兴趣圈』，回复『Java』获取200本优质电子书</p>
      
  </div>

  <div class="site-nav-toggle">
    <button>
      <span class="btn-bar"></span>
      <span class="btn-bar"></span>
      <span class="btn-bar"></span>
    </button>
  </div>
</div>

<nav class="site-nav">
  

  
    <ul id="menu" class="menu">
      
        
        <li class="menu-item menu-item-home">
          <a href="/" rel="section">
            
              <i class="menu-item-icon fa fa-fw fa-home"></i> <br />
            
            首页
          </a>
        </li>
      
        
        <li class="menu-item menu-item-categories">
          <a href="/categories/" rel="section">
            
              <i class="menu-item-icon fa fa-fw fa-question-circle"></i> <br />
            
            分类
          </a>
        </li>
      
        
        <li class="menu-item menu-item-archives">
          <a href="/archives/" rel="section">
            
              <i class="menu-item-icon fa fa-fw fa-question-circle"></i> <br />
            
            归档
          </a>
        </li>
      

      
    </ul>
  

  
</nav>



 </div>
    </header>

    <main id="main" class="main">
      <div class="main-inner">
        <div class="content-wrap">
          <div id="content" class="content">
            

  <div id="posts" class="posts-expand">
    

  

  
  
  

  <article class="post post-type-normal" itemscope itemtype="http://schema.org/Article">
  
  
  
  <div class="post-block">
    <link itemprop="mainEntityOfPage" href="https://www.codingw.net/posts/e8f03b64.html">

    <span hidden itemprop="author" itemscope itemtype="http://schema.org/Person">
      <meta itemprop="name" content="">
      <meta itemprop="description" content="">
      <meta itemprop="image" content="/images/avatar.gif">
    </span>

    <span hidden itemprop="publisher" itemscope itemtype="http://schema.org/Organization">
      <meta itemprop="name" content="中间件兴趣圈">
    </span>

    
      <header class="post-header">

        
        
          <h1 class="post-title" itemprop="name headline">源码分析RocketMQ消息轨迹</h1>
        

        <div class="post-meta">
          <span class="post-time">
            
              <span class="post-meta-item-icon">
                <i class="fa fa-calendar-o"></i>
              </span>
              
                <span class="post-meta-item-text">发表于</span>
              
              <time title="创建于" itemprop="dateCreated datePublished" datetime="2020-12-15T22:01:35+08:00">
                2020-12-15
              </time>
            

            

            
          </span>

          
            <span class="post-category" >
            
              <span class="post-meta-divider">|</span>
            
              <span class="post-meta-item-icon">
                <i class="fa fa-folder-o"></i>
              </span>
              
                <span class="post-meta-item-text">分类于</span>
              
              
                <span itemprop="about" itemscope itemtype="http://schema.org/Thing">
                  <a href="/categories/rocketmq/" itemprop="url" rel="index">
                    <span itemprop="name">rocketmq</span>
                  </a>
                </span>

                
                
              
            </span>
          

          
            
          

          
          
             <span id="/posts/e8f03b64.html" class="leancloud_visitors" data-flag-title="源码分析RocketMQ消息轨迹">
               <span class="post-meta-divider">|</span>
               <span class="post-meta-item-icon">
                 <i class="fa fa-eye"></i>
               </span>
               
                 <span class="post-meta-item-text">阅读次数&#58;</span>
               
                 <span class="leancloud-visitors-count"></span>
             </span>
          

          
            <span class="post-meta-divider">|</span>
            <span class="page-pv"><i class="fa fa-file-o"></i>
            <span class="busuanzi-value" id="busuanzi_value_page_pv" ></span>次
            </span>
          

          

          

        </div>
      </header>
    

    
    
    
    <div class="post-body" itemprop="articleBody">

      
      

      
        <div id="vip-container"><p>本文沿着<a target="_blank" rel="noopener" href="https://blog.csdn.net/prestigeding/article/details/95922489">《RocketMQ消息轨迹-设计篇》</a>的思路，从如下3个方面对其源码进行解读：</p>
<ol>
<li>发送消息轨迹</li>
<li>消息轨迹格式</li>
<li>存储消息轨迹数据</li>
</ol>
<h2 id="1、发送消息轨迹流程"><a href="#1、发送消息轨迹流程" class="headerlink" title="1、发送消息轨迹流程"></a>1、发送消息轨迹流程</h2><p>首先我们来看一下在消息发送端如何启用消息轨迹，示例代码如下：</p>
<span id="more"></span>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">TraceProducer</span> </span>&#123;</span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">void</span> <span class="title">main</span><span class="params">(String[] args)</span> <span class="keyword">throws</span> MQClientException, InterruptedException </span>&#123;</span><br><span class="line">        DefaultMQProducer producer = <span class="keyword">new</span> DefaultMQProducer(<span class="string">&quot;ProducerGroupName&quot;</span>,<span class="keyword">true</span>);      <span class="comment">// @1</span></span><br><span class="line">        producer.setNamesrvAddr(<span class="string">&quot;127.0.0.1:9876&quot;</span>);</span><br><span class="line">        producer.start();</span><br><span class="line">        <span class="keyword">for</span> (<span class="keyword">int</span> i = <span class="number">0</span>; i &lt; <span class="number">10</span>; i++)</span><br><span class="line">            <span class="keyword">try</span> &#123;</span><br><span class="line">                &#123;</span><br><span class="line">                    Message msg = <span class="keyword">new</span> Message(<span class="string">&quot;TopicTest&quot;</span>,</span><br><span class="line">                        <span class="string">&quot;TagA&quot;</span>,</span><br><span class="line">                        <span class="string">&quot;OrderID188&quot;</span>,</span><br><span class="line">                        <span class="string">&quot;Hello world&quot;</span>.getBytes(RemotingHelper.DEFAULT_CHARSET));</span><br><span class="line">                    SendResult sendResult = producer.send(msg);</span><br><span class="line">                    System.out.printf(<span class="string">&quot;%s%n&quot;</span>, sendResult);</span><br><span class="line">                &#125;</span><br><span class="line"></span><br><span class="line">            &#125; <span class="keyword">catch</span> (Exception e) &#123;</span><br><span class="line">                e.printStackTrace();</span><br><span class="line">            &#125;</span><br><span class="line">        producer.shutdown();</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<p>从上述代码可以看出其关键点是在创建DefaultMQProducer时指定开启消息轨迹跟踪。我们不妨浏览一下DefaultMQProducer与启用消息轨迹相关的构造函数：</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br></pre></td><td class="code"><pre><span class="line"><span class="function"><span class="keyword">public</span> <span class="title">DefaultMQProducer</span><span class="params">(<span class="keyword">final</span> String producerGroup, <span class="keyword">boolean</span> enableMsgTrace)</span></span></span><br><span class="line"><span class="function"><span class="keyword">public</span> <span class="title">DefaultMQProducer</span><span class="params">(<span class="keyword">final</span> String producerGroup, <span class="keyword">boolean</span> enableMsgTrace, <span class="keyword">final</span> String customizedTraceTopic)</span></span></span><br></pre></td></tr></table></figure>
<p>参数如下：</p>
<ul>
<li>String producerGroup<br>生产者所属组名。</li>
<li>boolean enableMsgTrace<br>是否开启跟踪消息轨迹，默认为false。</li>
<li>String customizedTraceTopic<br>如果开启消息轨迹跟踪，用来存储消息轨迹数据所属的主题名称，默认为：RMQ_SYS_TRACE_TOPIC。</li>
</ul>
<h3 id="1-1-DefaultMQProducer构造函数"><a href="#1-1-DefaultMQProducer构造函数" class="headerlink" title="1.1 DefaultMQProducer构造函数"></a>1.1 DefaultMQProducer构造函数</h3><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br></pre></td><td class="code"><pre><span class="line"><span class="function"><span class="keyword">public</span> <span class="title">DefaultMQProducer</span><span class="params">(<span class="keyword">final</span> String producerGroup, RPCHook rpcHook, <span class="keyword">boolean</span> enableMsgTrace,<span class="keyword">final</span> String customizedTraceTopic)</span> </span>&#123;      <span class="comment">// @1</span></span><br><span class="line">    <span class="keyword">this</span>.producerGroup = producerGroup;</span><br><span class="line">    defaultMQProducerImpl = <span class="keyword">new</span> DefaultMQProducerImpl(<span class="keyword">this</span>, rpcHook);</span><br><span class="line">    <span class="comment">//if client open the message trace feature</span></span><br><span class="line">    <span class="keyword">if</span> (enableMsgTrace) &#123;                                                                                                                                                                                            <span class="comment">// @2</span></span><br><span class="line">        <span class="keyword">try</span> &#123;</span><br><span class="line">            AsyncTraceDispatcher dispatcher = <span class="keyword">new</span> AsyncTraceDispatcher(customizedTraceTopic, rpcHook);                                                         </span><br><span class="line">            dispatcher.setHostProducer(<span class="keyword">this</span>.getDefaultMQProducerImpl());</span><br><span class="line">            traceDispatcher = dispatcher;</span><br><span class="line">            <span class="keyword">this</span>.getDefaultMQProducerImpl().registerSendMessageHook(</span><br><span class="line">                <span class="keyword">new</span> SendMessageTraceHookImpl(traceDispatcher));                                                                                                                             <span class="comment">// @3</span></span><br><span class="line">        &#125; <span class="keyword">catch</span> (Throwable e) &#123;</span><br><span class="line">            log.error(<span class="string">&quot;system mqtrace hook init failed ,maybe can&#x27;t send msg trace data&quot;</span>);</span><br><span class="line">        &#125;</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<p>代码@1：首先介绍一下其局部变量。</p>
<ul>
<li>String producerGroup<br>生产者所属组。</li>
<li>RPCHook rpcHook<br>生产者发送钩子函数。</li>
<li>boolean enableMsgTrace<br>是否开启消息轨迹跟踪。</li>
<li>String customizedTraceTopic<br>定制用于存储消息轨迹的数据。</li>
</ul>
<p>代码@2：用来构建AsyncTraceDispatcher，看其名：异步转发消息轨迹数据，稍后重点关注。</p>
<p>代码@3：构建SendMessageTraceHookImpl对象，并使用AsyncTraceDispatcher用来异步转发。</p>
<h3 id="1-2-SendMessageTraceHookImpl钩子函数"><a href="#1-2-SendMessageTraceHookImpl钩子函数" class="headerlink" title="1.2 SendMessageTraceHookImpl钩子函数"></a>1.2 SendMessageTraceHookImpl钩子函数</h3><h4 id="1-2-1-SendMessageTraceHookImpl类图"><a href="#1-2-1-SendMessageTraceHookImpl类图" class="headerlink" title="1.2.1 SendMessageTraceHookImpl类图"></a>1.2.1 SendMessageTraceHookImpl类图</h4><p><img src="https://img-blog.csdnimg.cn/20190803203518449.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ByZXN0aWdlZGluZw==,size_16,color_FFFFFF,t_70" alt="在这里插入图片描述"></p>
<ol>
<li>SendMessageHook<br>消息发送钩子函数，用于在消息发送之前、发送之后执行一定的业务逻辑，是记录消息轨迹的最佳扩展点。</li>
<li>TraceDispatcher<br>消息轨迹转发处理器，其默认实现类AsyncTraceDispatcher，异步实现消息轨迹数据的发送。下面对其属性做一个简单的介绍：<ul>
<li>int queueSize<br>异步转发，队列长度，默认为2048，当前版本不能修改。</li>
<li>int batchSize<br>批量消息条数，消息轨迹一次消息发送请求包含的数据条数，默认为100，当前版本不能修改。</li>
<li>int maxMsgSize<br>消息轨迹一次发送的最大消息大小，默认为128K，当前版本不能修改。</li>
<li>DefaultMQProducer traceProducer<br>  用来发送消息轨迹的消息发送者。</li>
<li>ThreadPoolExecutor traceExecuter<br>线程池，用来异步执行消息发送。</li>
<li>AtomicLong discardCount<br>记录丢弃的消息个数。</li>
<li>Thread worker<br>woker线程，主要负责从追加队列中获取一批待发送的消息轨迹数据，提交到线程池中执行。</li>
<li>ArrayBlockingQueue&lt; TraceContext&gt; traceContextQueue<br>消息轨迹TraceContext队列，用来存放待发送到服务端的消息。</li>
<li>ArrayBlockingQueue&lt; Runnable&gt; appenderQueue<br>线程池内部队列，默认长度1024。</li>
<li>DefaultMQPushConsumerImpl hostConsumer<br>消费者信息，记录消息消费时的轨迹信息。</li>
<li>String traceTopicName<br>用于跟踪消息轨迹的topic名称。</li>
</ul>
</li>
</ol>
<h4 id="1-2-2-源码分析SendMessageTraceHookImpl"><a href="#1-2-2-源码分析SendMessageTraceHookImpl" class="headerlink" title="1.2.2 源码分析SendMessageTraceHookImpl"></a>1.2.2 源码分析SendMessageTraceHookImpl</h4><h5 id="1-2-2-1-sendMessageBefore"><a href="#1-2-2-1-sendMessageBefore" class="headerlink" title="1.2.2.1 sendMessageBefore"></a>1.2.2.1 sendMessageBefore</h5><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br></pre></td><td class="code"><pre><span class="line"><span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">sendMessageBefore</span><span class="params">(SendMessageContext context)</span> </span>&#123; </span><br><span class="line">    <span class="comment">//if it is message trace data,then it doesn&#x27;t recorded</span></span><br><span class="line">    <span class="keyword">if</span> (context == <span class="keyword">null</span> || context.getMessage().getTopic().startsWith(((AsyncTraceDispatcher) localDispatcher).getTraceTopicName())) &#123;   <span class="comment">// @1</span></span><br><span class="line">        <span class="keyword">return</span>;</span><br><span class="line">    &#125;</span><br><span class="line">    <span class="comment">//build the context content of TuxeTraceContext</span></span><br><span class="line">    TraceContext tuxeContext = <span class="keyword">new</span> TraceContext();</span><br><span class="line">    tuxeContext.setTraceBeans(<span class="keyword">new</span> ArrayList&lt;TraceBean&gt;(<span class="number">1</span>));</span><br><span class="line">    context.setMqTraceContext(tuxeContext);</span><br><span class="line">    tuxeContext.setTraceType(TraceType.Pub);</span><br><span class="line">    tuxeContext.setGroupName(context.getProducerGroup());                                                                                                                       <span class="comment">// @2</span></span><br><span class="line">    <span class="comment">//build the data bean object of message trace</span></span><br><span class="line">    TraceBean traceBean = <span class="keyword">new</span> TraceBean();                                                                                                                                                <span class="comment">// @3</span></span><br><span class="line">    traceBean.setTopic(context.getMessage().getTopic());</span><br><span class="line">    traceBean.setTags(context.getMessage().getTags());</span><br><span class="line">    traceBean.setKeys(context.getMessage().getKeys());</span><br><span class="line">    traceBean.setStoreHost(context.getBrokerAddr());</span><br><span class="line">    traceBean.setBodyLength(context.getMessage().getBody().length);</span><br><span class="line">    traceBean.setMsgType(context.getMsgType());</span><br><span class="line">    tuxeContext.getTraceBeans().add(traceBean);</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<p>代码@1：如果topic主题为消息轨迹的Topic，直接返回。</p>
<p>代码@2：在消息发送上下文中，设置用来跟踪消息轨迹的上下环境，里面主要包含一个TraceBean集合、追踪类型（TraceType.Pub）与生产者所属的组。</p>
<p>代码@3：构建一条跟踪消息，用TraceBean来表示，记录原消息的topic、tags、keys、发送到broker地址、消息体长度等消息。</p>
<p>从上文看出，sendMessageBefore主要的用途就是在消息发送的时候，先准备一部分消息跟踪日志，存储在发送上下文环境中，此时并不会发送消息轨迹数据。</p>
<h5 id="1-2-2-2-sendMessageAfter"><a href="#1-2-2-2-sendMessageAfter" class="headerlink" title="1.2.2.2 sendMessageAfter"></a>1.2.2.2 sendMessageAfter</h5><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br></pre></td><td class="code"><pre><span class="line"><span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">sendMessageAfter</span><span class="params">(SendMessageContext context)</span> </span>&#123;</span><br><span class="line">    <span class="comment">//if it is message trace data,then it doesn&#x27;t recorded</span></span><br><span class="line">    <span class="keyword">if</span> (context == <span class="keyword">null</span> || context.getMessage().getTopic().startsWith(((AsyncTraceDispatcher) localDispatcher).getTraceTopicName())     <span class="comment">// @1</span></span><br><span class="line">        || context.getMqTraceContext() == <span class="keyword">null</span>) &#123;</span><br><span class="line">        <span class="keyword">return</span>;</span><br><span class="line">    &#125;</span><br><span class="line">    <span class="keyword">if</span> (context.getSendResult() == <span class="keyword">null</span>) &#123;</span><br><span class="line">        <span class="keyword">return</span>;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="keyword">if</span> (context.getSendResult().getRegionId() == <span class="keyword">null</span></span><br><span class="line">        || !context.getSendResult().isTraceOn()) &#123;</span><br><span class="line">        <span class="comment">// if switch is false,skip it</span></span><br><span class="line">        <span class="keyword">return</span>;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    TraceContext tuxeContext = (TraceContext) context.getMqTraceContext();</span><br><span class="line">    TraceBean traceBean = tuxeContext.getTraceBeans().get(<span class="number">0</span>);                                                                                                <span class="comment">// @2</span></span><br><span class="line">    <span class="keyword">int</span> costTime = (<span class="keyword">int</span>) ((System.currentTimeMillis() - tuxeContext.getTimeStamp()) / tuxeContext.getTraceBeans().size());     <span class="comment">// @3</span></span><br><span class="line">    tuxeContext.setCostTime(costTime);                                                                                                                                      <span class="comment">// @4</span></span><br><span class="line">    <span class="keyword">if</span> (context.getSendResult().getSendStatus().equals(SendStatus.SEND_OK)) &#123;                                                                    </span><br><span class="line">        tuxeContext.setSuccess(<span class="keyword">true</span>);</span><br><span class="line">    &#125; <span class="keyword">else</span> &#123;</span><br><span class="line">        tuxeContext.setSuccess(<span class="keyword">false</span>);</span><br><span class="line">    &#125;</span><br><span class="line">    tuxeContext.setRegionId(context.getSendResult().getRegionId());                                                                                      </span><br><span class="line">    traceBean.setMsgId(context.getSendResult().getMsgId());</span><br><span class="line">    traceBean.setOffsetMsgId(context.getSendResult().getOffsetMsgId());</span><br><span class="line">    traceBean.setStoreTime(tuxeContext.getTimeStamp() + costTime / <span class="number">2</span>);</span><br><span class="line">    localDispatcher.append(tuxeContext);                                                                                                                                   <span class="comment">// @5</span></span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<p>代码@1：如果topic主题为消息轨迹的Topic，直接返回。</p>
<p>代码@2：从MqTraceContext中获取跟踪的TraceBean，虽然设计成List结构体，但在消息发送场景，这里的数据永远只有一条，及时是批量发送也不例外。</p>
<p>代码@3：获取消息发送到收到响应结果的耗时。</p>
<p>代码@4：设置costTime(耗时)、success(是否发送成功)、regionId(发送到broker所在的分区)、msgId(消息ID，全局唯一)、offsetMsgId(消息物理偏移量，如果是批量消息，则是最后一条消息的物理偏移量)、storeTime，这里使用的是(客户端发送时间 + 二分之一的耗时)来表示消息的存储时间，这里是一个估值。</p>
<p>代码@5：将需要跟踪的信息通过TraceDispatcher转发到Broker服务器。其代码如下：</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br></pre></td><td class="code"><pre><span class="line"><span class="function"><span class="keyword">public</span> <span class="keyword">boolean</span> <span class="title">append</span><span class="params">(<span class="keyword">final</span> Object ctx)</span> </span>&#123;</span><br><span class="line">    <span class="keyword">boolean</span> result = traceContextQueue.offer((TraceContext) ctx);</span><br><span class="line">    <span class="keyword">if</span> (!result) &#123;</span><br><span class="line">        log.info(<span class="string">&quot;buffer full&quot;</span> + discardCount.incrementAndGet() + <span class="string">&quot; ,context is &quot;</span> + ctx);</span><br><span class="line">    &#125;</span><br><span class="line">    <span class="keyword">return</span> result;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<p>这里一个非常关键的点是offer方法的使用，当队列无法容纳新的元素时会立即返回false，并不会阻塞。</p>
<p>接下来将目光转向TraceDispatcher的实现。</p>
<h3 id="1-3-TraceDispatcher实现原理"><a href="#1-3-TraceDispatcher实现原理" class="headerlink" title="1.3 TraceDispatcher实现原理"></a>1.3 TraceDispatcher实现原理</h3><p>TraceDispatcher，用于客户端消息轨迹数据转发到Broker，其默认实现类：AsyncTraceDispatcher。</p>
<h4 id="1-3-1-TraceDispatcher构造函数"><a href="#1-3-1-TraceDispatcher构造函数" class="headerlink" title="1.3.1 TraceDispatcher构造函数"></a>1.3.1 TraceDispatcher构造函数</h4><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br></pre></td><td class="code"><pre><span class="line"><span class="function"><span class="keyword">public</span> <span class="title">AsyncTraceDispatcher</span><span class="params">(String traceTopicName, RPCHook rpcHook)</span> <span class="keyword">throws</span> MQClientException </span>&#123;    </span><br><span class="line">    <span class="comment">// queueSize is greater than or equal to the n power of 2 of value</span></span><br><span class="line">    <span class="keyword">this</span>.queueSize = <span class="number">2048</span>;</span><br><span class="line">    <span class="keyword">this</span>.batchSize = <span class="number">100</span>;</span><br><span class="line">    <span class="keyword">this</span>.maxMsgSize = <span class="number">128000</span>;                                        </span><br><span class="line">    <span class="keyword">this</span>.discardCount = <span class="keyword">new</span> AtomicLong(<span class="number">0L</span>);         </span><br><span class="line">    <span class="keyword">this</span>.traceContextQueue = <span class="keyword">new</span> ArrayBlockingQueue&lt;TraceContext&gt;(<span class="number">1024</span>);</span><br><span class="line">    <span class="keyword">this</span>.appenderQueue = <span class="keyword">new</span> ArrayBlockingQueue&lt;Runnable&gt;(queueSize);</span><br><span class="line">    <span class="keyword">if</span> (!UtilAll.isBlank(traceTopicName)) &#123;</span><br><span class="line">        <span class="keyword">this</span>.traceTopicName = traceTopicName;</span><br><span class="line">    &#125; <span class="keyword">else</span> &#123;</span><br><span class="line">        <span class="keyword">this</span>.traceTopicName = MixAll.RMQ_SYS_TRACE_TOPIC;</span><br><span class="line">    &#125;                   <span class="comment">// @1</span></span><br><span class="line">    <span class="keyword">this</span>.traceExecuter = <span class="keyword">new</span> ThreadPoolExecutor(<span class="comment">// :</span></span><br><span class="line">        <span class="number">10</span>, <span class="comment">//</span></span><br><span class="line">        <span class="number">20</span>, <span class="comment">//</span></span><br><span class="line">        <span class="number">1000</span> * <span class="number">60</span>, <span class="comment">//</span></span><br><span class="line">        TimeUnit.MILLISECONDS, <span class="comment">//</span></span><br><span class="line">        <span class="keyword">this</span>.appenderQueue, <span class="comment">//</span></span><br><span class="line">        <span class="keyword">new</span> ThreadFactoryImpl(<span class="string">&quot;MQTraceSendThread_&quot;</span>));</span><br><span class="line">    traceProducer = getAndCreateTraceProducer(rpcHook);      <span class="comment">// @2</span></span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<p>代码@1：初始化核心属性，该版本这些值都是“固化”的，用户无法修改。</p>
<ul>
<li>queueSize<br>队列长度，默认为2048，异步线程池能够积压的消息轨迹数量。</li>
<li>batchSize<br>一次向Broker批量发送的消息条数，默认为100.</li>
<li>maxMsgSize<br>向Broker汇报消息轨迹时，消息体的总大小不能超过该值，默认为128k。</li>
<li>discardCount<br>整个运行过程中，丢弃的消息轨迹数据，这里要说明一点的是，如果消息TPS发送过大，异步转发线程处理不过来时，会主动丢弃消息轨迹数据。</li>
<li>traceContextQueue<br>traceContext积压队列，客户端(消息发送、消息消费者)在收到处理结果后，将消息轨迹提交到噶队列中，则会立即返回。</li>
<li>appenderQueue<br>提交到Broker线程池中队列。</li>
<li>traceTopicName<br>用于接收消息轨迹的Topic，默认为RMQ_SYS_TRANS_HALF_TOPIC。</li>
<li>traceExecuter<br>用于发送到Broker服务的异步线程池，核心线程数默认为10，最大线程池为20，队列堆积长度2048，线程名称：MQTraceSendThread_。、</li>
<li>traceProducer<br>发送消息轨迹的Producer。</li>
</ul>
<p>代码@2：调用getAndCreateTraceProducer方法创建用于发送消息轨迹的Producer(消息发送者)，下面详细介绍一下其实现。</p>
<h4 id="1-3-2-getAndCreateTraceProducer详解"><a href="#1-3-2-getAndCreateTraceProducer详解" class="headerlink" title="1.3.2 getAndCreateTraceProducer详解"></a>1.3.2 getAndCreateTraceProducer详解</h4><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br></pre></td><td class="code"><pre><span class="line"><span class="function"><span class="keyword">private</span> DefaultMQProducer <span class="title">getAndCreateTraceProducer</span><span class="params">(RPCHook rpcHook)</span> </span>&#123;</span><br><span class="line">        DefaultMQProducer traceProducerInstance = <span class="keyword">this</span>.traceProducer;</span><br><span class="line">        <span class="keyword">if</span> (traceProducerInstance == <span class="keyword">null</span>) &#123;  <span class="comment">//@1</span></span><br><span class="line">            traceProducerInstance = <span class="keyword">new</span> DefaultMQProducer(rpcHook);</span><br><span class="line">            traceProducerInstance.setProducerGroup(TraceConstants.GROUP_NAME);</span><br><span class="line">            traceProducerInstance.setSendMsgTimeout(<span class="number">5000</span>);</span><br><span class="line">            traceProducerInstance.setVipChannelEnabled(<span class="keyword">false</span>);</span><br><span class="line">            <span class="comment">// The max size of message is 128K</span></span><br><span class="line">            traceProducerInstance.setMaxMessageSize(maxMsgSize - <span class="number">10</span> * <span class="number">1000</span>);</span><br><span class="line">        &#125;</span><br><span class="line">        <span class="keyword">return</span> traceProducerInstance;</span><br><span class="line">    &#125;</span><br></pre></td></tr></table></figure>
<p>代码@1：如果还未建立发送者，则创建用于发送消息轨迹的消息发送者，其GroupName为：_INNER_TRACE_PRODUCER，消息发送超时时间5s，最大允许发送消息大小118K。</p>
<h4 id="1-3-3-start"><a href="#1-3-3-start" class="headerlink" title="1.3.3 start"></a>1.3.3 start</h4><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br></pre></td><td class="code"><pre><span class="line"><span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">start</span><span class="params">(String nameSrvAddr)</span> <span class="keyword">throws</span> MQClientException </span>&#123;</span><br><span class="line">    <span class="keyword">if</span> (isStarted.compareAndSet(<span class="keyword">false</span>, <span class="keyword">true</span>)) &#123;     <span class="comment">// @1</span></span><br><span class="line">        traceProducer.setNamesrvAddr(nameSrvAddr);</span><br><span class="line">        traceProducer.setInstanceName(TRACE_INSTANCE_NAME + <span class="string">&quot;_&quot;</span> + nameSrvAddr);</span><br><span class="line">        traceProducer.start();</span><br><span class="line">    &#125;</span><br><span class="line">    <span class="keyword">this</span>.worker = <span class="keyword">new</span> Thread(<span class="keyword">new</span> AsyncRunnable(), <span class="string">&quot;MQ-AsyncTraceDispatcher-Thread-&quot;</span> + dispatcherId);   <span class="comment">// @2</span></span><br><span class="line">    <span class="keyword">this</span>.worker.setDaemon(<span class="keyword">true</span>);</span><br><span class="line">    <span class="keyword">this</span>.worker.start();                                                                                   </span><br><span class="line">    <span class="keyword">this</span>.registerShutDownHook();</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<p>开始启动，其调用的时机为启动DefaultMQProducer时，如果启用跟踪消息轨迹，则调用之。</p>
<p>代码@1：如果用于发送消息轨迹的发送者没有启动，则设置nameserver地址，并启动着。</p>
<p>代码@2：启动一个线程，用于执行AsyncRunnable任务，接下来将重点介绍。</p>
<h4 id="1-3-4-AsyncRunnable"><a href="#1-3-4-AsyncRunnable" class="headerlink" title="1.3.4 AsyncRunnable"></a>1.3.4 AsyncRunnable</h4><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br></pre></td><td class="code"><pre><span class="line"><span class="class"><span class="keyword">class</span> <span class="title">AsyncRunnable</span> <span class="keyword">implements</span> <span class="title">Runnable</span> </span>&#123;</span><br><span class="line">         <span class="keyword">private</span> <span class="keyword">boolean</span> stopped;</span><br><span class="line">	<span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">run</span><span class="params">()</span> </span>&#123;</span><br><span class="line">        <span class="keyword">while</span> (!stopped) &#123;</span><br><span class="line">            List&lt;TraceContext&gt; contexts = <span class="keyword">new</span> ArrayList&lt;TraceContext&gt;(batchSize);     <span class="comment">// @1</span></span><br><span class="line">            <span class="keyword">for</span> (<span class="keyword">int</span> i = <span class="number">0</span>; i &lt; batchSize; i++) &#123;</span><br><span class="line">                TraceContext context = <span class="keyword">null</span>;</span><br><span class="line">                <span class="keyword">try</span> &#123;</span><br><span class="line">                    <span class="comment">//get trace data element from blocking Queue — traceContextQueue</span></span><br><span class="line">                    context = traceContextQueue.poll(<span class="number">5</span>, TimeUnit.MILLISECONDS);        <span class="comment">// @2</span></span><br><span class="line">                &#125; <span class="keyword">catch</span> (InterruptedException e) &#123;</span><br><span class="line">                &#125;</span><br><span class="line">                <span class="keyword">if</span> (context != <span class="keyword">null</span>) &#123;</span><br><span class="line">                    contexts.add(context);</span><br><span class="line">                &#125; <span class="keyword">else</span> &#123;</span><br><span class="line">                    <span class="keyword">break</span>;</span><br><span class="line">                &#125;</span><br><span class="line">            &#125;</span><br><span class="line">            <span class="keyword">if</span> (contexts.size() &gt; <span class="number">0</span>) &#123;                                                                               :</span><br><span class="line">                AsyncAppenderRequest request = <span class="keyword">new</span> AsyncAppenderRequest(contexts);  <span class="comment">// @3</span></span><br><span class="line">                traceExecuter.submit(request);                                                               </span><br><span class="line">            &#125; <span class="keyword">else</span> <span class="keyword">if</span> (AsyncTraceDispatcher.<span class="keyword">this</span>.stopped) &#123;</span><br><span class="line">                <span class="keyword">this</span>.stopped = <span class="keyword">true</span>;</span><br><span class="line">            &#125;</span><br><span class="line">        &#125;</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<p>代码@1：构建待提交消息跟踪Bean，每次最多发送batchSize，默认为100条。</p>
<p>代码@2：从traceContextQueue中取出一个待提交的TraceContext，设置超时时间为5s，即如何该队列中没有待提交的TraceContext，则最多等待5s。</p>
<p>代码@3：向线程池中提交任务AsyncAppenderRequest。</p>
<h4 id="1-3-5-AsyncAppenderRequest-sendTraceData"><a href="#1-3-5-AsyncAppenderRequest-sendTraceData" class="headerlink" title="1.3.5 AsyncAppenderRequest#sendTraceData"></a>1.3.5 AsyncAppenderRequest#sendTraceData</h4><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br></pre></td><td class="code"><pre><span class="line"><span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">sendTraceData</span><span class="params">(List&lt;TraceContext&gt; contextList)</span> </span>&#123;</span><br><span class="line">    Map&lt;String, List&lt;TraceTransferBean&gt;&gt; transBeanMap = <span class="keyword">new</span> HashMap&lt;String, List&lt;TraceTransferBean&gt;&gt;();</span><br><span class="line">    <span class="keyword">for</span> (TraceContext context : contextList) &#123;        <span class="comment">//@1</span></span><br><span class="line">        <span class="keyword">if</span> (context.getTraceBeans().isEmpty()) &#123;</span><br><span class="line">            <span class="keyword">continue</span>;</span><br><span class="line">        &#125;</span><br><span class="line">        <span class="comment">// Topic value corresponding to original message entity content</span></span><br><span class="line">        String topic = context.getTraceBeans().get(<span class="number">0</span>).getTopic();     <span class="comment">// @2</span></span><br><span class="line">        <span class="comment">// Use  original message entity&#x27;s topic as key</span></span><br><span class="line">        String key = topic;</span><br><span class="line">        List&lt;TraceTransferBean&gt; transBeanList = transBeanMap.get(key);</span><br><span class="line">        <span class="keyword">if</span> (transBeanList == <span class="keyword">null</span>) &#123;</span><br><span class="line">            transBeanList = <span class="keyword">new</span> ArrayList&lt;TraceTransferBean&gt;();</span><br><span class="line">            transBeanMap.put(key, transBeanList);</span><br><span class="line">        &#125;</span><br><span class="line">        TraceTransferBean traceData = TraceDataEncoder.encoderFromContextBean(context);    <span class="comment">// @3</span></span><br><span class="line">        transBeanList.add(traceData);</span><br><span class="line">    &#125;</span><br><span class="line">    <span class="keyword">for</span> (Map.Entry&lt;String, List&lt;TraceTransferBean&gt;&gt; entry : transBeanMap.entrySet()) &#123;       <span class="comment">// @4</span></span><br><span class="line">        flushData(entry.getValue());</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<p>代码@1：遍历收集的消息轨迹数据。</p>
<p>代码@2：获取存储消息轨迹的Topic。</p>
<p>代码@3：对TraceContext进行编码，这里是消息轨迹的传输数据，稍后对其详细看一下，了解其上传的格式。</p>
<p>代码@4：将编码后的数据发送到Broker服务器。</p>
<h4 id="1-3-6-TraceDataEncoder-encoderFromContextBean"><a href="#1-3-6-TraceDataEncoder-encoderFromContextBean" class="headerlink" title="1.3.6 TraceDataEncoder#encoderFromContextBean"></a>1.3.6 TraceDataEncoder#encoderFromContextBean</h4><p>根据消息轨迹跟踪类型，其格式会有一些不一样，下面分别来介绍其合适。</p>
<h5 id="1-3-6-1-PUB-消息发送"><a href="#1-3-6-1-PUB-消息发送" class="headerlink" title="1.3.6.1 PUB(消息发送)"></a>1.3.6.1 PUB(消息发送)</h5><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">case</span> Pub: &#123;</span><br><span class="line">    TraceBean bean = ctx.getTraceBeans().get(<span class="number">0</span>);</span><br><span class="line">    <span class="comment">//append the content of context and traceBean to transferBean&#x27;s TransData</span></span><br><span class="line">    sb.append(ctx.getTraceType()).append(TraceConstants.CONTENT_SPLITOR)<span class="comment">//</span></span><br><span class="line">      .append(ctx.getTimeStamp()).append(TraceConstants.CONTENT_SPLITOR)<span class="comment">//</span></span><br><span class="line">      .append(ctx.getRegionId()).append(TraceConstants.CONTENT_SPLITOR)<span class="comment">//</span></span><br><span class="line">      .append(ctx.getGroupName()).append(TraceConstants.CONTENT_SPLITOR)<span class="comment">//</span></span><br><span class="line">      .append(bean.getTopic()).append(TraceConstants.CONTENT_SPLITOR)<span class="comment">//</span></span><br><span class="line">      .append(bean.getMsgId()).append(TraceConstants.CONTENT_SPLITOR)<span class="comment">//</span></span><br><span class="line">      .append(bean.getTags()).append(TraceConstants.CONTENT_SPLITOR)<span class="comment">//</span></span><br><span class="line">      .append(bean.getKeys()).append(TraceConstants.CONTENT_SPLITOR)<span class="comment">//</span></span><br><span class="line">      .append(bean.getStoreHost()).append(TraceConstants.CONTENT_SPLITOR)<span class="comment">//</span></span><br><span class="line">      .append(bean.getBodyLength()).append(TraceConstants.CONTENT_SPLITOR)<span class="comment">//</span></span><br><span class="line">      .append(ctx.getCostTime()).append(TraceConstants.CONTENT_SPLITOR)<span class="comment">//</span></span><br><span class="line">      .append(bean.getMsgType().ordinal()).append(TraceConstants.CONTENT_SPLITOR)<span class="comment">//</span></span><br><span class="line">      .append(bean.getOffsetMsgId()).append(TraceConstants.CONTENT_SPLITOR)<span class="comment">//</span></span><br><span class="line">     .append(ctx.isSuccess()).append(TraceConstants.FIELD_SPLITOR);</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<p>消息轨迹数据的协议使用字符串拼接，字段的分隔符号为1，整个数据以2结尾，感觉这个设计还是有点“不可思议”，为什么不直接使用json协议呢？</p>
<h5 id="1-3-6-2-SubBefore-消息消费之前"><a href="#1-3-6-2-SubBefore-消息消费之前" class="headerlink" title="1.3.6.2 SubBefore(消息消费之前)"></a>1.3.6.2 SubBefore(消息消费之前)</h5><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">for</span> (TraceBean bean : ctx.getTraceBeans()) &#123;</span><br><span class="line">    sb.append(ctx.getTraceType()).append(TraceConstants.CONTENT_SPLITOR)<span class="comment">//</span></span><br><span class="line">      .append(ctx.getTimeStamp()).append(TraceConstants.CONTENT_SPLITOR)<span class="comment">//</span></span><br><span class="line">      .append(ctx.getRegionId()).append(TraceConstants.CONTENT_SPLITOR)<span class="comment">//</span></span><br><span class="line">      .append(ctx.getGroupName()).append(TraceConstants.CONTENT_SPLITOR)<span class="comment">//</span></span><br><span class="line">      .append(ctx.getRequestId()).append(TraceConstants.CONTENT_SPLITOR)<span class="comment">//</span></span><br><span class="line">      .append(bean.getMsgId()).append(TraceConstants.CONTENT_SPLITOR)<span class="comment">//</span></span><br><span class="line">      .append(bean.getRetryTimes()).append(TraceConstants.CONTENT_SPLITOR)<span class="comment">//</span></span><br><span class="line">      .append(bean.getKeys()).append(TraceConstants.FIELD_SPLITOR);<span class="comment">//</span></span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<p>轨迹就是按照上述顺序拼接而成，各个字段使用1分隔，每一条记录使用2结尾。</p>
<h5 id="1-3-2-3-SubAfter（消息消费后）"><a href="#1-3-2-3-SubAfter（消息消费后）" class="headerlink" title="1.3.2.3 SubAfter（消息消费后）"></a>1.3.2.3 SubAfter（消息消费后）</h5><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">case</span> SubAfter: &#123;</span><br><span class="line">    <span class="keyword">for</span> (TraceBean bean : ctx.getTraceBeans()) &#123;</span><br><span class="line">        sb.append(ctx.getTraceType()).append(TraceConstants.CONTENT_SPLITOR)<span class="comment">//</span></span><br><span class="line">          .append(ctx.getRequestId()).append(TraceConstants.CONTENT_SPLITOR)<span class="comment">//</span></span><br><span class="line">          .append(bean.getMsgId()).append(TraceConstants.CONTENT_SPLITOR)<span class="comment">//</span></span><br><span class="line">          .append(ctx.getCostTime()).append(TraceConstants.CONTENT_SPLITOR)<span class="comment">//</span></span><br><span class="line">          .append(ctx.isSuccess()).append(TraceConstants.CONTENT_SPLITOR)<span class="comment">//</span></span><br><span class="line">          .append(bean.getKeys()).append(TraceConstants.CONTENT_SPLITOR)<span class="comment">//</span></span><br><span class="line">          .append(ctx.getContextCode()).append(TraceConstants.FIELD_SPLITOR);</span><br><span class="line">        &#125;</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<p>格式编码一样，就不重复多说。</p>
<p>经过上面的源码跟踪，消息发送端的消息轨迹跟踪流程、消息轨迹数据编码协议就清晰了，接下来我们使用一张序列图来结束本部分的讲解。<br><img src="https://img-blog.csdnimg.cn/20190803204514547.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ByZXN0aWdlZGluZw==,size_16,color_FFFFFF,t_70" alt="在这里插入图片描述"><br>其实行文至此，只关注了消息发送的消息轨迹跟踪，消息消费的轨迹跟踪又是如何呢？其实现原理其实是一样的，就是在消息消费前后执行特定的钩子函数，其实现类为ConsumeMessageTraceHookImpl，由于其实现与消息发送的思路类似，故就不详细介绍了。</p>
<h2 id="2、-消息轨迹数据如何存储"><a href="#2、-消息轨迹数据如何存储" class="headerlink" title="2、 消息轨迹数据如何存储"></a>2、 消息轨迹数据如何存储</h2><p>其实从上面的分析，我们已经得知，RocketMQ的消息轨迹数据存储在到Broker上，那消息轨迹的主题名如何指定？其路由信息又怎么分配才好呢？是每台Broker上都创建还是只在其中某台上创建呢？RocketMQ支持系统默认与自定义消息轨迹的主题。</p>
<h3 id="2-1-使用系统默认的主题名称"><a href="#2-1-使用系统默认的主题名称" class="headerlink" title="2.1 使用系统默认的主题名称"></a>2.1 使用系统默认的主题名称</h3><p>RocketMQ默认的消息轨迹主题为：RMQ_SYS_TRACE_TOPIC，那该Topic需要手工创建吗？其路由信息呢？</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br></pre></td><td class="code"><pre><span class="line">&#123;</span><br><span class="line">    <span class="keyword">if</span> (<span class="keyword">this</span>.brokerController.getBrokerConfig().isTraceTopicEnable()) &#123;    <span class="comment">// @1</span></span><br><span class="line">        String topic = <span class="keyword">this</span>.brokerController.getBrokerConfig().getMsgTraceTopicName();</span><br><span class="line">        TopicConfig topicConfig = <span class="keyword">new</span> TopicConfig(topic);</span><br><span class="line">        <span class="keyword">this</span>.systemTopicList.add(topic);</span><br><span class="line">        topicConfig.setReadQueueNums(<span class="number">1</span>);                                              <span class="comment">// @2</span></span><br><span class="line">        topicConfig.setWriteQueueNums(<span class="number">1</span>);</span><br><span class="line">        <span class="keyword">this</span>.topicConfigTable.put(topicConfig.getTopicName(), topicConfig);</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<p>上述代码出自TopicConfigManager的构造函数，在Broker启动的时候会创建topicConfigManager对象，用来管理topic的路由信息。</p>
<p>代码@1：如果Broker开启了消息轨迹跟踪(traceTopicEnable=true)时，会自动创建默认消息轨迹的topic路由信息，注意其读写队列数为1。</p>
<h3 id="2-2-用户自定义消息轨迹主题"><a href="#2-2-用户自定义消息轨迹主题" class="headerlink" title="2.2 用户自定义消息轨迹主题"></a>2.2 用户自定义消息轨迹主题</h3><p>在创建消息发送者、消息消费者时，可以显示的指定消息轨迹的Topic，例如：</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br></pre></td><td class="code"><pre><span class="line"><span class="function"><span class="keyword">public</span> <span class="title">DefaultMQProducer</span><span class="params">(<span class="keyword">final</span> String producerGroup, RPCHook rpcHook, <span class="keyword">boolean</span> enableMsgTrace,<span class="keyword">final</span> String customizedTraceTopic)</span></span></span><br><span class="line"><span class="function"></span></span><br><span class="line"><span class="function"><span class="keyword">public</span> <span class="title">DefaultMQPushConsumer</span><span class="params">(<span class="keyword">final</span> String consumerGroup, RPCHook rpcHook,</span></span></span><br><span class="line"><span class="function"><span class="params">        AllocateMessageQueueStrategy allocateMessageQueueStrategy, <span class="keyword">boolean</span> enableMsgTrace, <span class="keyword">final</span> String customizedTraceTopic)</span></span></span><br></pre></td></tr></table></figure>
<p>通过customizedTraceTopic来指定消息轨迹Topic。</p>
<blockquote>
<p>温馨提示：通常在生产环境上，将不会开启自动创建主题，故需要RocketMQ运维管理人员提前创建好Topic。</p>
</blockquote>
<p>好了，本文就介绍到这里了，本文详细介绍了RocktMQ消息轨迹的实现原理，下一篇，我们将进入到多副本的学习中。</p>
</div>

			<script src="https://my.openwrite.cn/js/readmore.js" type="text/javascript"></script>
			<script>
			var isMobile = navigator.userAgent.match(/(phone|pad|pod|iPhone|iPod|ios|iPad|Android|Mobile|BlackBerry|IEMobile|MQQBrowser|JUC|Fennec|wOSBrowser|BrowserNG|WebOS|Symbian|Windows Phone)/i);
			if (!isMobile) {
			    var btw = new BTWPlugin();
			    btw.init({
			        "id": "vip-container",
			        "blogId": "18019-1573088808868-542",
			        "name": "中间件兴趣圈",
			        "qrcode": "https://img-blog.csdnimg.cn/20190314214003962.jpg",
			        "keyword": "more"
			    });
			}
			</script>
		
      
    </div>
    
    
    

    

    

    

    <footer class="post-footer">
      

      
      
      

      
        <div class="post-nav">
          <div class="post-nav-next post-nav-item">
            
              <a href="/posts/e3c7511d.html" rel="next" title="AtomicIntegerArray源码分析与感悟.">
                <i class="fa fa-chevron-left"></i> AtomicIntegerArray源码分析与感悟.
              </a>
            
          </div>

          <span class="post-nav-divider"></span>

          <div class="post-nav-prev post-nav-item">
            
              <a href="/posts/a6ee7996.html" rel="prev" title="源码分析RocketMQ ACL实现机制">
                源码分析RocketMQ ACL实现机制 <i class="fa fa-chevron-right"></i>
              </a>
            
          </div>
        </div>
      

      
      
    </footer>
  </div>
  
  
  
  </article>



    <div class="post-spread">
      
    </div>
  </div>


          </div>
          


          

  



        </div>
        
          
  
  <div class="sidebar-toggle">
    <div class="sidebar-toggle-line-wrap">
      <span class="sidebar-toggle-line sidebar-toggle-line-first"></span>
      <span class="sidebar-toggle-line sidebar-toggle-line-middle"></span>
      <span class="sidebar-toggle-line sidebar-toggle-line-last"></span>
    </div>
  </div>

  <aside id="sidebar" class="sidebar">
    
    <div class="sidebar-inner">

      

      
        <ul class="sidebar-nav motion-element">
          <li class="sidebar-nav-toc sidebar-nav-active" data-target="post-toc-wrap">
            文章目录
          </li>
          <li class="sidebar-nav-overview" data-target="site-overview-wrap">
            站点概览
          </li>
        </ul>
      

      <section class="site-overview-wrap sidebar-panel">
        <div class="site-overview">
          <div class="site-author motion-element" itemprop="author" itemscope itemtype="http://schema.org/Person">
            
              <p class="site-author-name" itemprop="name"></p>
              <p class="site-description motion-element" itemprop="description"></p>
          </div>

          <nav class="site-state motion-element">

            
              <div class="site-state-item site-state-posts">
              
                <a href="/archives/">
              
                  <span class="site-state-item-count">139</span>
                  <span class="site-state-item-name">日志</span>
                </a>
              </div>
            

            
              
              
              <div class="site-state-item site-state-categories">
                <a href="/categories/index.html">
                  <span class="site-state-item-count">18</span>
                  <span class="site-state-item-name">分类</span>
                </a>
              </div>
            

            

          </nav>

          

          

          
          

          
          

          

        </div>
      </section>

      
      <!--noindex-->
        <section class="post-toc-wrap motion-element sidebar-panel sidebar-panel-active">
          <div class="post-toc">

            
              
            

            
              <div class="post-toc-content"><ol class="nav"><li class="nav-item nav-level-2"><a class="nav-link" href="#1%E3%80%81%E5%8F%91%E9%80%81%E6%B6%88%E6%81%AF%E8%BD%A8%E8%BF%B9%E6%B5%81%E7%A8%8B"><span class="nav-number">1.</span> <span class="nav-text">1、发送消息轨迹流程</span></a><ol class="nav-child"><li class="nav-item nav-level-3"><a class="nav-link" href="#1-1-DefaultMQProducer%E6%9E%84%E9%80%A0%E5%87%BD%E6%95%B0"><span class="nav-number">1.1.</span> <span class="nav-text">1.1 DefaultMQProducer构造函数</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#1-2-SendMessageTraceHookImpl%E9%92%A9%E5%AD%90%E5%87%BD%E6%95%B0"><span class="nav-number">1.2.</span> <span class="nav-text">1.2 SendMessageTraceHookImpl钩子函数</span></a><ol class="nav-child"><li class="nav-item nav-level-4"><a class="nav-link" href="#1-2-1-SendMessageTraceHookImpl%E7%B1%BB%E5%9B%BE"><span class="nav-number">1.2.1.</span> <span class="nav-text">1.2.1 SendMessageTraceHookImpl类图</span></a></li><li class="nav-item nav-level-4"><a class="nav-link" href="#1-2-2-%E6%BA%90%E7%A0%81%E5%88%86%E6%9E%90SendMessageTraceHookImpl"><span class="nav-number">1.2.2.</span> <span class="nav-text">1.2.2 源码分析SendMessageTraceHookImpl</span></a><ol class="nav-child"><li class="nav-item nav-level-5"><a class="nav-link" href="#1-2-2-1-sendMessageBefore"><span class="nav-number">1.2.2.1.</span> <span class="nav-text">1.2.2.1 sendMessageBefore</span></a></li><li class="nav-item nav-level-5"><a class="nav-link" href="#1-2-2-2-sendMessageAfter"><span class="nav-number">1.2.2.2.</span> <span class="nav-text">1.2.2.2 sendMessageAfter</span></a></li></ol></li></ol></li><li class="nav-item nav-level-3"><a class="nav-link" href="#1-3-TraceDispatcher%E5%AE%9E%E7%8E%B0%E5%8E%9F%E7%90%86"><span class="nav-number">1.3.</span> <span class="nav-text">1.3 TraceDispatcher实现原理</span></a><ol class="nav-child"><li class="nav-item nav-level-4"><a class="nav-link" href="#1-3-1-TraceDispatcher%E6%9E%84%E9%80%A0%E5%87%BD%E6%95%B0"><span class="nav-number">1.3.1.</span> <span class="nav-text">1.3.1 TraceDispatcher构造函数</span></a></li><li class="nav-item nav-level-4"><a class="nav-link" href="#1-3-2-getAndCreateTraceProducer%E8%AF%A6%E8%A7%A3"><span class="nav-number">1.3.2.</span> <span class="nav-text">1.3.2 getAndCreateTraceProducer详解</span></a></li><li class="nav-item nav-level-4"><a class="nav-link" href="#1-3-3-start"><span class="nav-number">1.3.3.</span> <span class="nav-text">1.3.3 start</span></a></li><li class="nav-item nav-level-4"><a class="nav-link" href="#1-3-4-AsyncRunnable"><span class="nav-number">1.3.4.</span> <span class="nav-text">1.3.4 AsyncRunnable</span></a></li><li class="nav-item nav-level-4"><a class="nav-link" href="#1-3-5-AsyncAppenderRequest-sendTraceData"><span class="nav-number">1.3.5.</span> <span class="nav-text">1.3.5 AsyncAppenderRequest#sendTraceData</span></a></li><li class="nav-item nav-level-4"><a class="nav-link" href="#1-3-6-TraceDataEncoder-encoderFromContextBean"><span class="nav-number">1.3.6.</span> <span class="nav-text">1.3.6 TraceDataEncoder#encoderFromContextBean</span></a><ol class="nav-child"><li class="nav-item nav-level-5"><a class="nav-link" href="#1-3-6-1-PUB-%E6%B6%88%E6%81%AF%E5%8F%91%E9%80%81"><span class="nav-number">1.3.6.1.</span> <span class="nav-text">1.3.6.1 PUB(消息发送)</span></a></li><li class="nav-item nav-level-5"><a class="nav-link" href="#1-3-6-2-SubBefore-%E6%B6%88%E6%81%AF%E6%B6%88%E8%B4%B9%E4%B9%8B%E5%89%8D"><span class="nav-number">1.3.6.2.</span> <span class="nav-text">1.3.6.2 SubBefore(消息消费之前)</span></a></li><li class="nav-item nav-level-5"><a class="nav-link" href="#1-3-2-3-SubAfter%EF%BC%88%E6%B6%88%E6%81%AF%E6%B6%88%E8%B4%B9%E5%90%8E%EF%BC%89"><span class="nav-number">1.3.6.3.</span> <span class="nav-text">1.3.2.3 SubAfter（消息消费后）</span></a></li></ol></li></ol></li></ol></li><li class="nav-item nav-level-2"><a class="nav-link" href="#2%E3%80%81-%E6%B6%88%E6%81%AF%E8%BD%A8%E8%BF%B9%E6%95%B0%E6%8D%AE%E5%A6%82%E4%BD%95%E5%AD%98%E5%82%A8"><span class="nav-number">2.</span> <span class="nav-text">2、 消息轨迹数据如何存储</span></a><ol class="nav-child"><li class="nav-item nav-level-3"><a class="nav-link" href="#2-1-%E4%BD%BF%E7%94%A8%E7%B3%BB%E7%BB%9F%E9%BB%98%E8%AE%A4%E7%9A%84%E4%B8%BB%E9%A2%98%E5%90%8D%E7%A7%B0"><span class="nav-number">2.1.</span> <span class="nav-text">2.1 使用系统默认的主题名称</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#2-2-%E7%94%A8%E6%88%B7%E8%87%AA%E5%AE%9A%E4%B9%89%E6%B6%88%E6%81%AF%E8%BD%A8%E8%BF%B9%E4%B8%BB%E9%A2%98"><span class="nav-number">2.2.</span> <span class="nav-text">2.2 用户自定义消息轨迹主题</span></a></li></ol></li></ol></div>
            

          </div>
        </section>
      <!--/noindex-->
      

      

    </div>
  </aside>


        
      </div>
    </main>

    <footer id="footer" class="footer">
      <div class="footer-inner">
        <div class="copyright">&copy; <span itemprop="copyrightYear">2021</span>
  <span class="with-love">
    <i class="fa fa-user"></i>
  </span>
  <span class="author" itemprop="copyrightHolder">中间件兴趣圈</span>

  
</div>


  <div class="powered-by">由 <a class="theme-link" target="_blank" href="https://hexo.io">Hexo</a> 强力驱动</div>



  <span class="post-meta-divider">|</span>



  <div class="theme-info">主题 &mdash; <a class="theme-link" target="_blank" href="https://github.com/iissnan/hexo-theme-next">NexT.Muse</a> v5.1.4</div>




        
<div class="busuanzi-count">
  <script async src="https://dn-lbstatics.qbox.me/busuanzi/2.3/busuanzi.pure.mini.js"></script>

  
    <span class="site-uv">
      <i class="fa fa-user"></i>
      <span class="busuanzi-value" id="busuanzi_value_site_uv"></span>
      
    </span>
  

  
    <span class="site-pv">
      <i class="fa fa-eye"></i>
      <span class="busuanzi-value" id="busuanzi_value_site_pv"></span>
      
    </span>
  
</div>








        
      </div>
    </footer>

    
      <div class="back-to-top">
        <i class="fa fa-arrow-up"></i>
        
      </div>
    

    

  </div>

  

<script type="text/javascript">
  if (Object.prototype.toString.call(window.Promise) !== '[object Function]') {
    window.Promise = null;
  }
</script>









  












  
  
    <script type="text/javascript" src="/lib/jquery/index.js?v=2.1.3"></script>
  

  
  
    <script type="text/javascript" src="/lib/fastclick/lib/fastclick.min.js?v=1.0.6"></script>
  

  
  
    <script type="text/javascript" src="/lib/jquery_lazyload/jquery.lazyload.js?v=1.9.7"></script>
  

  
  
    <script type="text/javascript" src="/lib/velocity/velocity.min.js?v=1.2.1"></script>
  

  
  
    <script type="text/javascript" src="/lib/velocity/velocity.ui.min.js?v=1.2.1"></script>
  

  
  
    <script type="text/javascript" src="/lib/fancybox/source/jquery.fancybox.pack.js?v=2.1.5"></script>
  


  


  <script type="text/javascript" src="/js/src/utils.js?v=5.1.4"></script>

  <script type="text/javascript" src="/js/src/motion.js?v=5.1.4"></script>



  
  

  
  <script type="text/javascript" src="/js/src/scrollspy.js?v=5.1.4"></script>
<script type="text/javascript" src="/js/src/post-details.js?v=5.1.4"></script>



  


  <script type="text/javascript" src="/js/src/bootstrap.js?v=5.1.4"></script>



  


  




	





  





  












  





  

  
  <script src="https://cdn1.lncld.net/static/js/av-core-mini-0.6.4.js"></script>
  <script>AV.initialize("NNEhOL0iOcflg8f1U3HUqiCq-gzGzoHsz", "7kSmkbbb3DktmHALlShDsBUF");</script>
  <script>
    function showTime(Counter) {
      var query = new AV.Query(Counter);
      var entries = [];
      var $visitors = $(".leancloud_visitors");

      $visitors.each(function () {
        entries.push( $(this).attr("id").trim() );
      });

      query.containedIn('url', entries);
      query.find()
        .done(function (results) {
          var COUNT_CONTAINER_REF = '.leancloud-visitors-count';

          if (results.length === 0) {
            $visitors.find(COUNT_CONTAINER_REF).text(0);
            return;
          }

          for (var i = 0; i < results.length; i++) {
            var item = results[i];
            var url = item.get('url');
            var time = item.get('time');
            var element = document.getElementById(url);

            $(element).find(COUNT_CONTAINER_REF).text(time);
          }
          for(var i = 0; i < entries.length; i++) {
            var url = entries[i];
            var element = document.getElementById(url);
            var countSpan = $(element).find(COUNT_CONTAINER_REF);
            if( countSpan.text() == '') {
              countSpan.text(0);
            }
          }
        })
        .fail(function (object, error) {
          console.log("Error: " + error.code + " " + error.message);
        });
    }

    function addCount(Counter) {
      var $visitors = $(".leancloud_visitors");
      var url = $visitors.attr('id').trim();
      var title = $visitors.attr('data-flag-title').trim();
      var query = new AV.Query(Counter);

      query.equalTo("url", url);
      query.find({
        success: function(results) {
          if (results.length > 0) {
            var counter = results[0];
            counter.fetchWhenSave(true);
            counter.increment("time");
            counter.save(null, {
              success: function(counter) {
                var $element = $(document.getElementById(url));
                $element.find('.leancloud-visitors-count').text(counter.get('time'));
              },
              error: function(counter, error) {
                console.log('Failed to save Visitor num, with error message: ' + error.message);
              }
            });
          } else {
            var newcounter = new Counter();
            /* Set ACL */
            var acl = new AV.ACL();
            acl.setPublicReadAccess(true);
            acl.setPublicWriteAccess(true);
            newcounter.setACL(acl);
            /* End Set ACL */
            newcounter.set("title", title);
            newcounter.set("url", url);
            newcounter.set("time", 1);
            newcounter.save(null, {
              success: function(newcounter) {
                var $element = $(document.getElementById(url));
                $element.find('.leancloud-visitors-count').text(newcounter.get('time'));
              },
              error: function(newcounter, error) {
                console.log('Failed to create');
              }
            });
          }
        },
        error: function(error) {
          console.log('Error:' + error.code + " " + error.message);
        }
      });
    }

    $(function() {
      var Counter = AV.Object.extend("Counter");
      if ($('.leancloud_visitors').length == 1) {
        addCount(Counter);
      } else if ($('.post-title-link').length > 1) {
        showTime(Counter);
      }
    });
  </script>



  

  

  
  

  

  

  

</body>
</html>
